package org.dlx.drpc.core.server;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.timeout.ReadTimeoutHandler;
import org.dlx.drpc.core.common.*;

public class RPCServer implements IRPCServer
{
    private String ip;
    private int port;
    private int workerThreads; // 用于业务处理的计算线程
    private RPCDispatcher dispatcher;
    private IRPCEncoder rpcEncoder;
    private IRPCDecoder rpcDecoder;
    private int readTimeoutSeconds;
    private ChannelFuture closeFuture;
    private EventLoopGroup group;

    //readTimeoutSeconds 超过指定时间未读取到客户端数据,则主动关闭客户端链接
    public RPCServer(String ip, int port, int ioThreads, int workerThreads, int readTimeoutSeconds, IRPCEncoder en, IRPCDecoder de) {
        this.ip = ip;
        this.port = port;
        this.workerThreads = workerThreads;
        this.dispatcher = new RPCDispatcher(this.workerThreads);
        this.rpcEncoder = en;
        this.rpcDecoder = de;
        this.readTimeoutSeconds = readTimeoutSeconds;
        this.group = new NioEventLoopGroup(ioThreads);
    }
    @Override
    public RPCServer register(Class<?> clazz, IRPCHandler handler) {
        this.dispatcher.register(clazz, handler);
        return this;
    }
    @Override
    public String getAddress(){
        return ip + ":" + Integer.toString(this.port);
    }
    @Override
    public RPCServer start() throws Exception{
        ServerBootstrap b = new ServerBootstrap();
        b.group(group).channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>()
                    {
                        @Override
                        public void initChannel(SocketChannel ch) throws Exception{
                            ChannelPipeline pipe = ch.pipeline();
                            pipe.addLast(new ReadTimeoutHandler(readTimeoutSeconds));
                            //解密添加长度解析
                            pipe.addLast(new LengthFieldBasedFrameDecoder(65536, 0, 4, 0, 4));
                            //加密添加一个长度字段
                            pipe.addLast(new LengthFieldPrepender(4, false));
                            pipe.addLast(new MessageDecoder());//将bytebuf编码成message
                            pipe.addLast(new MessageEncoder());//将Message 编码成bytebuf
                            pipe.addLast(new RPCDecoder(rpcDecoder));
                            pipe.addLast(new RPCEncoder(rpcEncoder));
                            pipe.addLast(dispatcher);
                        }
                    }).option(ChannelOption.SO_BACKLOG, 100)
                    .option(ChannelOption.SO_REUSEADDR, true)// reuse addr，避免端口冲突
                    .option(ChannelOption.TCP_NODELAY, true)// 关闭小流合并，保证消息的及时性
                    .childOption(ChannelOption.SO_KEEPALIVE, true); // 长时间没动静的链接自动关闭
        ChannelFuture f = b.bind(this.ip, this.port).sync();
        closeFuture = f.channel().closeFuture();
        return this;
    }
    @Override
    public RPCServer working() throws Exception{
        closeFuture.sync();
        return this;
    }
    //清理资源
    @Override
    public RPCServer close(){
        group.shutdownGracefully();
        dispatcher.closeGracefully();
        return this;
    }

}
